package com.yujiahao;

/**
 * @Description: 还没跑通
 * @Author: yujiahao
 * @Date: 2023/8/2 18:42
 * @Version: 1.0
 */

import org.apache.flink.kinesis.shaded.com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
import org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.AmazonKinesis;
import org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.AmazonKinesisClientBuilder;
import org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream;
import org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
import org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.clientlibrary.lib.worker.Worker;


public class KinesisConsumeIRecordProcessorV2Test {

    public static void main(String[] args) {
        String streamName = "test-kds-api"; // 替换为您的Kinesis流名称
        String region = "us-east-2"; // 替换为您的Kinesis流所在的AWS区域

        String workerId = "worker-1"; // 替换为一个唯一的工作进程ID
        String appName = "KinesisConsumerApp"; // 替换为您的应用程序名称

        // 创建Kinesis客户端
        AmazonKinesis client = AmazonKinesisClientBuilder.standard()
                .withRegion(region)
                .withCredentials(new DefaultAWSCredentialsProviderChain())
                .build();

        // 创建Kinesis消费者配置
        KinesisClientLibConfiguration config = new KinesisClientLibConfiguration(appName, streamName,
                new DefaultAWSCredentialsProviderChain(), workerId)
                .withRegionName(region)
                .withInitialPositionInStream(InitialPositionInStream.TRIM_HORIZON); // 设置从最新数据开始消费

        // 创建消费者记录处理工厂
        ConsumerRecordProcessorFactory recordProcessorFactory = new ConsumerRecordProcessorFactory();

        // 创建Kinesis消费者Worker并启动
        Worker worker = new Worker.Builder()
                .recordProcessorFactory(recordProcessorFactory)
                .config(config)
                .kinesisClient(client)
                .build();

        worker.run();
    }
}

